package com.beesphere.products.throttling;

import com.beesphere.products.throttling.processor.async.AsyncProcessor;
import com.beesphere.products.throttling.processor.async.DelayAsyncProcessorSupport;

public class Throttler extends DelayAsyncProcessorSupport {
	
    protected UnitOfWorkCapacity capacity;
    
    private long requestCount;

    public Throttler (AsyncProcessor processor, UnitOfWorkCapacity capacity) {
        super (processor);
        this.capacity = capacity;
    }
    
    public boolean process (UnitOfWork unitOfWork, AsyncCallback asyncCallback) {
    	return super.process (unitOfWork, capacity.getTimeout (), asyncCallback);
    }

    protected void delay (UnitOfWork unitOfWork) throws Exception {
        long now = currentSystemTime();
        long startTimeMillis = capacity.getStartTimeMillis ();
        if (startTimeMillis == 0) {
            startTimeMillis = now;
        }
        if (now - startTimeMillis > capacity.getTimePeriodMillis ()) {
            // we're at the start of a new time period
            // so lets reset things
            requestCount = 1;
            startTimeMillis = now;
        } else {
            if (++requestCount > capacity.getMaximumRequestsPerPeriod ()) {
                // lets sleep until the start of the next time period
                long time = startTimeMillis + capacity.getTimePeriodMillis ();
                waitUntil (time, unitOfWork);
            }
        }
    }

	@Override
	protected boolean isRunAllowed () {
		return true;
	}

	public UnitOfWorkCapacity getCapacity () {
		return capacity;
	}

	public void setCapacity (UnitOfWorkCapacity capacity) {
		this.capacity = capacity;
	}
}
